package com.ky.pro.java.mq.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.jms.*;

@Service
@Slf4j
public class ActiveMqMessagingService implements MessagingService, DisposableBean {

    @Value("${activemq.broker.url}")
    String brokerUrl;

    private Session session;

    ActiveMQConnection conn;

    @PostConstruct
    public void init() throws JMSException {
        // 创建连接和会话
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUrl);
        conn = (ActiveMQConnection) factory.createConnection();
        conn.start();
        session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    }

    @Override
    public boolean sendToQueue(String queueId, String message) {
        try {
            MessageProducer producer = session.createProducer(new ActiveMQQueue(queueId));
            log.info("发送点对点消息： {}", message);
            producer.send(session.createTextMessage(message));
        } catch (JMSException e) {
            log.error("发送点对点消息失败", e);
            return false;
        }
        return true;
    }

    @Override
    public boolean publish(String topic, String message) {
        try {
            MessageProducer producer = session.createProducer(new ActiveMQTopic(topic));
            log.info("发送订阅消息： {}", message);
            producer.send(session.createTextMessage(message));
        } catch (JMSException e) {
            log.error("发送订阅消息失败", e);
            return false;
        }
        return true;
    }

    @Override
    public boolean receiveFromQueue(String queueId, IMessageListener listener) {
        try {
            MessageConsumer consumer = session.createConsumer(new ActiveMQQueue(queueId));
            consumer.setMessageListener((message -> {
                try {
                    listener.onMessage(((TextMessage) message).getText());
                } catch (JMSException e) {
                    log.error("接收消息发生异常", e);
                }
            }));
        } catch (JMSException e) {
            log.error("创建消息监听发生异常", e);
            return false;
        }
        return true;
    }

    @Override
    public boolean subscribe(String topic, IMessageListener listener) {
        try {
            MessageConsumer consumer = session.createConsumer(new ActiveMQTopic(topic));
            consumer.setMessageListener((message -> {
                try {
                    listener.onMessage(((TextMessage) message).getText());
                } catch (JMSException e) {
                    log.error("接收消息发生异常", e);
                }
            }));
        } catch (JMSException e) {
            log.error("创建消息监听发生异常", e);
            return false;
        }
        return true;
    }

    @Override
    public void destroy() throws Exception {
        session.close();
        conn.close();
    }
}
